fix(redis source): add reconnecting pubsub session for channel data_type - #25892
fix(redis source): add reconnecting pubsub session for channel data_type#25892artusmode wants to merge 8 commits into
data_type#25892Conversation
…_type`
- Rework channel source to maintain a pubsub session that auto-reconnects
and re-subscribes after Redis disconnects.
- Add shutdown-aware backoff and best-effort unsubscribe on exit.
- Log when the connection is re-established ("re-established and resubscribed").
Resolves vectordotdev#22615
Follow-up to the reconnecting pub/sub work: - Replace the ad-hoc `backoff_exponential` helper with the shared `common::backoff::ExponentialBackoff` (from_millis(2).factor(250), capped at 30s), matching the strategy used by other reconnecting sources such as `aws_s3`/`sqs`. Backoff resets on a successful connect. - Emit internal events for reconnection so operators can observe/alert: `RedisConnectionError` (increments `component_errors_total` with error_type=connection_failed) on connect/subscribe failure, and `RedisConnectionEstablished` (increments `connection_established_total`) on initial connect and recovery. - Unify the `list` data_type source (`list.rs`) onto the same shared `ExponentialBackoff`, removing its duplicate `backoff_exponential` helper and making its retry sleep shutdown-aware. - Add a changelog fragment. Co-authored-by: Gibran <gibran.zaman@finaccel.co>
|
All contributors have signed the CLA ✍️ ✅ |
|
🔗 Commit SHA: 0160cdc | Docs | Datadog PR Page | Give us feedback! |
There was a problem hiding this comment.
💡 Codex Review
Here are some automated review suggestions for this pull request.
Reviewed commit: aa85aeb2f1
ℹ️ About Codex in GitHub
Codex has been enabled to automatically review pull requests in this repo. Reviews are triggered when you
- Open a pull request for review
- Mark a draft as ready
- Comment "@codex review".
If Codex has suggestions, it will comment; otherwise it will react with 👍.
When you sign up for Codex through ChatGPT, Codex can also answer questions or update the PR, like "@codex address that feedback".
- Race `connect_and_subscribe` with the shutdown signal (biased select) so a connect against a black-holed endpoint can't block graceful shutdown until the force-shutdown deadline. - Drop the explicit `UNSUBSCRIBE` round trips on the shutdown/disconnect paths; dropping the pubsub connection closes it and Redis releases the subscription automatically, so shutdown no longer waits on a network round trip. - Keep the `list` source's retry backoff capped at 1s (its prior behavior) rather than the channel source's 30s cap, avoiding a recovery-latency regression for existing `data_type = "list"` users.
9f6f014 to
ac72174
Compare
|
I have read the CLA Document and I hereby sign the CLA |
There was a problem hiding this comment.
💡 Codex Review
Here are some automated review suggestions for this pull request.
Reviewed commit: ac72174894
ℹ️ About Codex in GitHub
Codex has been enabled to automatically review pull requests in this repo. Reviews are triggered when you
- Open a pull request for review
- Mark a draft as ready
- Comment "@codex review".
If Codex has suggestions, it will comment; otherwise it will react with 👍.
When you sign up for Codex through ChatGPT, Codex can also answer questions or update the PR, like "@codex address that feedback".
Second round of automated review feedback: - Do the initial connect + SUBSCRIBE during source build and fail fast on non-recoverable errors (bad auth/ACL/config) by returning the error, instead of entering the reconnect loop. An invalid configuration now surfaces at startup rather than looking like it started and only erroring at runtime. Transient (I/O) errors are still tolerated and handled by the reconnect loop. - Only reset the reconnect backoff after a session has actually delivered a message, not merely on a successful connect. A connection that is accepted and then immediately dropped (a flapping outage) now keeps backing off exponentially instead of hot-looping at the 500ms floor. - Classify permanent vs transient connect errors during reconnect too: a non-recoverable error stops the source rather than retrying forever.
There was a problem hiding this comment.
💡 Codex Review
Here are some automated review suggestions for this pull request.
Reviewed commit: 780538f5c0
ℹ️ About Codex in GitHub
Codex has been enabled to automatically review pull requests in this repo. Reviews are triggered when you
- Open a pull request for review
- Mark a draft as ready
- Comment "@codex review".
If Codex has suggestions, it will comment; otherwise it will react with 👍.
When you sign up for Codex through ChatGPT, Codex can also answer questions or update the PR, like "@codex address that feedback".
Use redis-rs's own `RetryMethod` classification instead of only treating `ErrorKind::IoError` as transient. A Redis node coming back from a restart may accept the TCP connection but reply to `SUBSCRIBE` with a server-side retryable error such as `LOADING`/`BUSYLOADING`/`TRYAGAIN`, which redis-rs reports as a non-I/O error. The previous check treated those as permanent and stopped the source, defeating the restart recovery this change is meant to provide. Now only `RetryMethod::NoRetry` (auth/ACL/config rejections) is fatal; all other errors, including those retryable server states, back off and reconnect.
There was a problem hiding this comment.
💡 Codex Review
Here are some automated review suggestions for this pull request.
Reviewed commit: 48254413e6
ℹ️ About Codex in GitHub
Codex has been enabled to automatically review pull requests in this repo. Reviews are triggered when you
- Open a pull request for review
- Mark a draft as ready
- Comment "@codex review".
If Codex has suggestions, it will comment; otherwise it will react with 👍.
When you sign up for Codex through ChatGPT, Codex can also answer questions or update the PR, like "@codex address that feedback".
… sessions Third round of automated review feedback: - redis-rs maps `ErrorKind::AuthenticationFailed` to `RetryMethod::Reconnect`, so the `RetryMethod`-based classifier treated bad credentials as transient and retried forever. Treat `AuthenticationFailed` as permanent explicitly so an invalid password/ACL fails the source build as intended. - Reset the reconnect backoff after a session has either delivered a message or stayed connected for a minimum healthy duration, rather than only on delivered data. A low-volume channel that stays healthy but quiet now clears a backoff that a previous flapping period drove up to the cap, so a later drop resubscribes promptly instead of waiting up to 30s.
There was a problem hiding this comment.
💡 Codex Review
Here are some automated review suggestions for this pull request.
Reviewed commit: de4b3cba8d
ℹ️ About Codex in GitHub
Codex has been enabled to automatically review pull requests in this repo. Reviews are triggered when you
- Open a pull request for review
- Mark a draft as ready
- Comment "@codex review".
If Codex has suggestions, it will comment; otherwise it will react with 👍.
When you sign up for Codex through ChatGPT, Codex can also answer questions or update the PR, like "@codex address that feedback".
Replace the error-classification approach with the source's original build-time semantics, which is both simpler and more robust: - The initial connect + SUBSCRIBE fails fast on ANY error (auth, TLS, ACL, unreachable, LOADING), surfacing misconfiguration at startup as the source did before reconnect support. This avoids fragile transient-vs-permanent classification: redis-rs reports TLS handshake failures and transient network timeouts both as IoError/RetryImmediately, so no kind/retry_method predicate can separate them. Removing `is_transient` also drops the auth special-case. - Once the source is running, every reconnect failure is retried; a permanent problem was already ruled out by the successful build-time connect. Failures are still recorded via RedisConnectionError for alerting. - Emit a new RedisConnectionDropped internal event (component_errors_total) when an established connection drops, so alerts fire even when the reconnect succeeds on the first attempt.
|
This supersedes #24100 (rebased, @gibranbadrul credited as co-author) and addresses all the automated review feedback. Verified locally with cargo check + cargo clippy -D warnings on --features sources-redis. Could a maintainer approve the workflows so CI can run? @pront, you reviewed the original approach on #24100 — would appreciate your eyes here when you have a moment. Thanks! |
@artusmode thanks. I will take a look. Can you help by resolving the open threads? Also, if they were valid findings, please spend a few seconds to react to them with 👍 (or 👎 ). |
data_typedata_type
`vdev check events` requires any internal event that increments `component_errors_total` to be named `*Error`. Rename `RedisConnectionDropped` -> `RedisConnectionDroppedError`.
|
Thanks @pront! A couple of things:
Could you re-approve the workflow run so CI can finish on the latest push? Appreciate the review! |
Supersedes #24100 (rebased onto master; original work by @gibranbadrul, preserved as the first commit). Closes #22615.
The
redissource withdata_type = "channel"currently stops receiving after any Redis connection drop and only recovers on a Vector restart. This adds a reconnect loop that re-connects and re-subscribes automatically, with exponential backoff.Addresses the review feedback on #24100:
common::backoff::ExponentialBackoff(from_millis(2).factor(250), capped at 30s), matchingaws_s3/sqs, instead of a bespoke helper. Backoff resets on a successful connect.RedisConnectionError→component_errors_total(error_type=connection_failed) on connect/subscribe failure, andRedisConnectionEstablished→connection_established_totalon initial connect and recovery.Also unifies the
listdata_type source onto the same sharedExponentialBackoff(removing its duplicatebackoff_exponential) and makes its retry sleep shutdown-aware.